#include "znx_single_flight.h"


static znx_single_flight_call_t *
znx_single_flight_call_create_helper(znx_single_flight_t *sf,
    znx_dispatch_event_t *dispatch_event,
    znx_dispatch_handler_t do_handler)
{
    znx_rbtree_node_t           *node;
    znx_single_flight_call_t    *call;
    znx_single_flight_call_t    target;
    znx_dispatch_event_t        *new_dispatch_event;

    target.master = dispatch_event;
    target.owner_sf = sf;
    node = znx_rbtree_lookup(&sf->tree, &target.tree_node);
    if (node == NULL) {
        call = znx_thread_calloc(sizeof(znx_single_flight_call_t));
        call->owner_sf = sf;
        znx_queue_init(&call->pending);

        if (sf->create_master == NULL) {
            call->master = dispatch_event;
            dispatch_event->is_master = 1;
            dispatch_event = NULL;
        } else {
            new_dispatch_event = sf->create_master(dispatch_event);
            call->master = new_dispatch_event;
            new_dispatch_event->is_master = 1;
        }

        znx_rbtree_insert(&sf->tree, &call->tree_node);

        if (dispatch_event == NULL) {
            return call;
        }

        znx_queue_insert_head(&call->pending, &dispatch_event->queue);

        do_handler(new_dispatch_event);

        return NULL;
    }

    call = (znx_single_flight_call_t *)node;
    znx_queue_insert_head(&call->pending, &dispatch_event->queue);
    return NULL;
}


void
znx_single_flight_do(znx_single_flight_t *sf,
    znx_dispatch_event_t *dispatch_event,
    znx_dispatch_handler_t do_handler)
{
    assert(current_thread_is_worker_thread());

    znx_single_flight_call_t *call;

    dispatch_event->setup_thread = current_thread_base;

    znx_spinlock_lock(&sf->lock);
    call = znx_single_flight_call_create_helper(sf,
        dispatch_event, do_handler);
    znx_spinlock_unlock(&sf->lock);

    // it's slaver
    if (call == NULL) {
        return;
    }

    // it's master.
    do_handler(dispatch_event);
}


znx_single_flight_call_t *
znx_single_flight_call_lookup_helper(znx_single_flight_t *sf,
    znx_dispatch_event_t *dispatch_event)
{
    znx_rbtree_node_t           *node;
    znx_single_flight_call_t    *call;
    znx_single_flight_call_t    target;

    target.master = dispatch_event;
    target.owner_sf = sf;
    node = znx_rbtree_lookup(&sf->tree, &target.tree_node);
    if (node == NULL) {
        return NULL;
    }

    call = (znx_single_flight_call_t *)node;
    return call;
}


znx_single_flight_call_t *
znx_single_flight_call_remove_helper(znx_single_flight_t *sf,
    znx_dispatch_event_t *dispatch_event)
{
    znx_single_flight_call_t    *call;

    call = znx_single_flight_call_lookup_helper(sf, dispatch_event);
    if (call == NULL) {
        ZNX_DEF_LOG_ERROR("not found master sf call");
        return NULL;
    }

    if (call->master != dispatch_event) {
        ZNX_DEF_LOG_ERROR("dispatch_event is not current call.master");
        return NULL;
    }

    znx_rbtree_delete(&sf->tree, &call->tree_node);
    return call;
}


void
znx_single_flight_notify(znx_single_flight_t *sf,
    znx_dispatch_event_t *dispatch_event,
    void *result)
{
    znx_queue_t                 *q;
    znx_single_flight_call_t    *call;
    znx_dispatch_event_t        *p_dispatch_event;

    znx_spinlock_lock(&sf->lock);
    call = znx_single_flight_call_remove_helper(sf, dispatch_event);
    znx_spinlock_unlock(&sf->lock);

    if (call == NULL) {
        return;
    }

    while (!znx_queue_empty(&call->pending)) {
        q = znx_queue_last(&call->pending);
        znx_queue_remove(q);

        p_dispatch_event = znx_queue_data(q, znx_dispatch_event_t, queue);
        if (sf->hold_result && result != NULL) {
            sf->hold_result(result, ZNX_FALSE);
        }
        p_dispatch_event->save_data_2 = result;
        znx_event_loop_post_event(p_dispatch_event->setup_thread, p_dispatch_event);
    }

    znx_thread_free(call);

    dispatch_event->save_data_2 = result;
    if (sf->hold_result && result != NULL) {
        sf->hold_result(result, ZNX_TRUE);
    }
    dispatch_event->handler(dispatch_event);
}


static znx_bool_t
znx_single_flight_cancel_helper(znx_single_flight_t *sf,
    znx_dispatch_event_t *dispatch_event)
{
    znx_single_flight_call_t    *call;
    call = znx_single_flight_call_lookup_helper(sf, dispatch_event);
    if (call == NULL) {
        return ZNX_FALSE;
    }

    if (call->master == dispatch_event) {
        return ZNX_FALSE;
    }

    if (dispatch_event->queue.prev == NULL) {
        return ZNX_FALSE;
    }

    znx_queue_remove(&dispatch_event->queue);

    return ZNX_TRUE;
}


znx_bool_t
znx_single_flight_cancel(znx_single_flight_t *sf,
	znx_dispatch_event_t *dispatch_event)
{
    znx_bool_t res;

    znx_spinlock_lock(&sf->lock);
    res = znx_single_flight_cancel_helper(sf, dispatch_event);
    znx_spinlock_unlock(&sf->lock);

    return res;
}
